/**通常在生产者中定义交换器和路由键，而在消费者中定义队列，并将队列绑定到交换器和相同的路由键
 * 1、Direct路由模式
 * 2、Topic主题工模式
 * 3、Fanout发布订阅模式
 */
package com.example.rabbitMQ;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class ExchangeDemo {

    private static final String ROUTING_KEY1 = "sms";
    private static final String ROUTING_KEY2 = "email";

    //-----------路由模式，按照routing-key发送和接收------------
    private static final String EXCHANGE_DIRECT = "direct_exchange";
    private static final String QUEUE_DIRECT1 = "direct_queue_1";
    private static final String QUEUE_DIRECT2 = "direct_queue_2";
    public static void directProduce() {
        try{
            Connection connection =   BasicDemo.getConnection();
            Channel channel = connection.createChannel();

            // 声明exchange，指定类型为topic
            channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT);
            String message = "注册成功！";
            //生产者发送消息，并且指定routing key 为：sms，只有短信服务能接收到消息
            channel.basicPublish(EXCHANGE_DIRECT, ROUTING_KEY2, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println("[(Direct)生产者]sent：" + message);
            channel.close();
            connection.close();
        }   catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void directConsume1() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_DIRECT1, false, false, false, null);

            //消费者将队列与交换机进行绑定
            channel.queueBind(QUEUE_DIRECT1, EXCHANGE_DIRECT, ROUTING_KEY1);
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) {
                    String msg = new String(body);
                    System.out.println("[(Direct)短信服务]received :" + msg);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            channel.basicConsume(QUEUE_DIRECT1, consumer);
            //等待回调函数执行完毕之后 关闭资源
            TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void directConsume2() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_DIRECT2, false, false, false, null);
            // 声明exchange，指定类型为direct
            channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT);
            //消费者将队列与交换机进行绑定
            channel.queueBind(QUEUE_DIRECT2, EXCHANGE_DIRECT, ROUTING_KEY2);
            channel.basicConsume(QUEUE_DIRECT2, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body){
                    String msg = new String(body);
                    System.out.println("[(Direct)邮件服务]received :" + msg);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //-----------主题模式（通配符），按照routing-key发送和接收------------
    private static final String EXCHANGE_TOPIC = "topic_exchange";
    private static final String QUEUE_TOPIC1 = "topic_queue_1";
    private static final String QUEUE_TOPIC2 = "topic_queue_2";
    public static void topicProduce() {
        try{
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            // 声明exchange，指定类型为topic
            channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC);
            String message = "这是一只行动迅速的橙色的兔子（Topic）";
            //生产者发送消息时，设置消息的Routing Key:"*/#"
            channel.basicPublish(EXCHANGE_TOPIC, "quick.orange.rabbit", null, message.getBytes());
            System.out.println("[Topic生产者]sent：" + message);
            channel.close();
            connection.close();
        }   catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void topicConsume1() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_TOPIC1, false, false, false, null);
            // 声明exchange，指定类型为topic
            channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC);
            //绑定队列到交换机，同时指定需要订阅的routing key。订阅所有的橙色动物，“*”：匹配一个单词
            channel.queueBind(QUEUE_TOPIC1, EXCHANGE_TOPIC, "*.orange.*",null);
            channel.basicConsume(QUEUE_TOPIC1, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                        throws IOException {
                    String msg = new String(body);
                    System.out.println("[Topic--1]received：" + msg);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void topicConsume2() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_TOPIC2, false, false, false, null);

            //绑定队列到交换机，同时指定需要订阅的routing key。订阅关于兔子以及懒惰动物的消息，“#”：匹配一个或多个单词
            channel.queueBind(QUEUE_TOPIC2, EXCHANGE_TOPIC, "*.*.rabbit");
            channel.queueBind(QUEUE_TOPIC2, EXCHANGE_TOPIC, "lazy.＃");

            channel.basicConsume(QUEUE_TOPIC2, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                        throws IOException {
                    String msg = new String(body);
                    System.out.println("[Topic--2]received：" + msg);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //-----------订阅模式，广播式发送和接收------------
    private static final String EXCHANGE_FANOUT = "fanout_exchange";
    private static final String QUEUE_FANOUT1 = "fanout_queue_1";
    private static final String QUEUE_FANOUT2 = "fanout_queue_2";
    public static void fanoutProduce() {
        try{
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();

            //声明Exchange，不再声明Queue
            channel.exchangeDeclare(EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT);
            String message = "注册成功！(Fanout)";
            //发送消息到Exchange，不再发送到Queue
            channel.basicPublish(EXCHANGE_FANOUT, "", null, message.getBytes());
            System.out.println("[(Fanout)生产者]sent：" + message);
            channel.close();
            connection.close();
        }   catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static void fanoutConsume1() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_FANOUT1, false, false, false, null);

            //消费者将队列与交换机进行绑定
            channel.queueBind(QUEUE_FANOUT1, EXCHANGE_FANOUT, "");
            //监听队列，自动返回完成
            channel.basicConsume(QUEUE_FANOUT1, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                        throws IOException {
                    String msg = new String(body);
                    System.out.println("[(Fanout)短信服务]received：" + msg);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fanoutConsume2() {
        try {
            Connection connection =  BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            //消费者声明自己的队列
            channel.queueDeclare(QUEUE_FANOUT2, false, false, false, null);
            //消费者将队列与交换机进行绑定
            channel.queueBind(QUEUE_FANOUT2, EXCHANGE_FANOUT, "");
            channel.basicConsume(QUEUE_FANOUT2, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                        throws IOException {
                    String msg = new String(body);
                    System.out.println("[(Fanout)邮件服务]received：" + msg);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
